package SQL_L

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

class Intro {
  @Test
  def rddIntro(): Unit = {
    val conf = new SparkConf().setMaster("local[6]").setAppName("Intro")
    val sc = new SparkContext(conf)
    sc.textFile("data/wordcount.txt")
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .collect()
      .foreach(println(_))
  }

  @Test
  def sqlIntro(): Unit = {
    val spark = new SparkSession.Builder()
      .appName("sqlIntro")
      .master("local[6]")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._
    val sourceRDD = spark.sparkContext.parallelize(Seq(Person("zhangsan", 19), Person("lisi", 21)))
    val personDS = sourceRDD.toDS()
    val resultDS = personDS.where('age > 20)
      .select('name)
      .as[String]
    resultDS.show()
  }

  @Test
  def dfIntro(): Unit = {
    val spark = new SparkSession.Builder()
      .appName("dfIntro")
      .master("local[6]")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._
    val sourceRDD = spark.sparkContext.parallelize(Seq(Person("zhangsan", 19), Person("lisi", 21)))
    val df = sourceRDD.toDF()
    //注册成一个临时表
    df.where('age > 20)
      .select('name)
      .show()
  }

  @Test
  def df2Intro(): Unit = {
    val spark = new SparkSession.Builder()
      .appName("df2Intro")
      .master("local[6]")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._
    //使用toDF创建dataframe
    val df = Seq(Person("zhangsan", 19), Person("lisi", 21)).toDF()
    //使用createDataFrame创建
    val df1 = spark.createDataFrame(Seq(Person("zhangsan", 19), Person("lisi", 21)))
    //使用read创建
    val df2 = spark.read.csv("csv文件")
    //注册成一个临时表
    df.createTempView("person")
    val resultdf = spark.sql("select name from person where age > 20")
    resultdf.show()
  }

  @Test
  def dataFrame2Test(): Unit = {
    //创建SparkSession
    val spark = new SparkSession.Builder()
      .master("local[6]")
      .appName("dataFrame3")
      .getOrCreate()
    //读取数据集
    import spark.implicits._
    val sourceDF = spark.read
      .option("header", true)
      .csv("data/BeijingPM20100101_20151231.csv")
    sourceDF.select('year, 'month, 'PM_Dongsi)
      .where('PM_Dongsi =!= "NA")
      .groupBy('year, 'month)
      .count()
      .show()
    //还可以直接使用sql语句来创建
    sourceDF.createOrReplaceTempView("pm")
    spark.sql("select year, month ,count(PM_Dongsi) from pm where PM_Dongsi != 'NA' group by year,month")
      .show()
    spark.stop()
  }

  @Test
  def dataSetIntro(): Unit = {
    //创建SparkSession
    val spark = new SparkSession.Builder()
      .master("local[6]")
      .appName("dataSetIntro")
      .getOrCreate()
    //导入隐式转换
    import spark.implicits._
    val sourceRDD = spark.sparkContext.parallelize(Seq(Person("zhangsan", 19), Person("lisi", 21)))
    val ds = sourceRDD.toDS()
    //处理
    //dataset支持RDD的API(强类型)
    ds.filter(item => item.age > 20).show()
    //也支持弱类型的API
    ds.filter('age > 20).show()
    ds.filter($"age" > 20).show()
    //可以直接编写sql表达式
    ds.filter("age > 20").show()

  }

  @Test
  def dataSet2Intro(): Unit = {
    //创建SparkSession
    val spark = new SparkSession.Builder()
      .master("local[6]")
      .appName("dataSet2Intro")
      .getOrCreate()
    //导入隐式转换
    import spark.implicits._
    val ds = spark.createDataset(Seq(Person("zhangsan", 19), Person("lisi", 21)))
    //处理
    //dataset支持RDD的API(强类型)
    ds.filter(item => item.age > 20).show()
    //也支持弱类型的API
    ds.filter('age > 20).show()
    ds.filter($"age" > 20).show()
    //可以直接编写sql表达式
    ds.filter("age > 20").show()

  }

  /**
   * dataFrame和dataSet的区别
   * dataFrame就是dataSet，DataFrame是dataSet的row泛型
   * dataFrame是弱类型，dataSet是强类型
   * dataFrame可以检测出运行时的错误
   * dataSet可以检测出运行和编译时的错误
   */
  @Test
  def rowTest(): Unit = {
    //Row如何创建，它是什么？
    val p = Person("zhangsan", 21

      /**
       * row和Person对象最大的区别就是row对象没有列名
       * row对象必须配合Schema对象，才会有列名
       */
    )
    val row = Row("zhangsan", 21)
    //从Row中获取数据
    row.getString(0)
    row.getInt(1)
    //ROW也可以作为样例类
    row match {
      case Row(name,age) => println(name,age)
    }
  }
}

case class Person(name: String, age: Int)